package com.gogant.spider.core;

import java.net.URL;
import java.net.URLConnection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * URL处理工作线程
 *
 * @param <Req>
 * @param <Resp>
 * @param <Conn>
 */
final class QueueWorker<Req extends Request, Resp extends Response, Conn extends URLConnection> implements Runnable {
    private static AtomicInteger workerCounter = new AtomicInteger();
    private static Object globalWorkerLock = new Object();
    private static long lastWorkTime;
    private QueueWorkerPool<Req, Resp, Conn> workerPool;
    private Spider<Req, Resp, Conn> spider;
    private Thread thread;
    private Context context;
    private boolean isWorking = false;

    QueueWorker(Spider<Req, Resp, Conn> spider, QueueWorkerPool<Req, Resp, Conn> workerPool) {
        this.spider = spider;
        this.workerPool = workerPool;
    }

    @SuppressWarnings("unchecked")
    private Connector<Spider<Req, Resp, Conn>, Req, Resp, Conn> createConnector(String protocol) throws SpiderException {
        ConnectorFactory<? extends Spider<Req, Resp, Conn>, Req, Resp, Conn> factory = workerPool.getConnectorFactory();
        if (factory == null) {
            throw new SpiderException("connector factory not exists");
        }
        Connector<Spider<Req, Resp, Conn>, Req, Resp, Conn> conn = (Connector<Spider<Req, Resp, Conn>, Req, Resp, Conn>) factory.createConnector(protocol);
        if (conn == null) {
            throw new SpiderException("factory[" + factory + "] return null connector");
        }
        return conn;
    }

    /**
     * 打开URL连接
     *
     * @param connector
     * @param req
     * @return
     * @throws SpiderException
     */
    private Conn openURLConnection(Connector<Spider<Req, Resp, Conn>, Req, Resp, Conn> connector, Request req) throws SpiderException {
        Conn conn = connector.openConnection(req.getURL(), spider.getProxy());
        if (conn == null) {
            throw new SpiderException("connector[" + connector + "] return null connection");
        }
        if (spider.getConnectTimeout() > 0) {
            conn.setConnectTimeout(spider.getConnectTimeout());
        }
        if (spider.getReadTimeout() > 0) {
            conn.setReadTimeout(spider.getReadTimeout());
        }
        Map<String, String> propertyMap = req.getHeaders();
        if (propertyMap != null && propertyMap.size() > 0) {
            for (Map.Entry<String, String> entry : propertyMap.entrySet()) {
                conn.addRequestProperty(entry.getKey(), entry.getValue());
            }
        }
        return conn;
    }

    /**
     * 匹配URLHandler，将URL交给指定的Handler进行处理
     *
     * @param context
     */
    private void handleFilter(Context context) {
        for (URLHandler<Req, Resp> handler : spider.getURLHandlerArray()) {
            if (handler.onFilter(context)) {
                context.addHandler(handler);
            }
        }
    }

    /**
     * 在发送请求前交给Handler进行处理
     *
     * @param req
     * @param context
     * @throws SpiderException
     */
    @SuppressWarnings("unchecked")
    private void handleRequest(Req req, Context context) throws SpiderException {
        for (RequestHandler<Req> requestHandler : spider.getGlobalRequestHandlerArray()) {
            requestHandler.onRequest(req);
        }
        if (context.hasHandler()) {
            for (URLHandler<?, ?> handler : context.getHandlerList()) {
                ((URLHandler<Req, Resp>) handler).onRequest(req);
            }
        }
    }

    /**
     * 从响应中提取URL
     *
     * @param resp
     * @param context
     * @throws SpiderException
     */
    private void handleURLPicker(Resp resp, Context context) throws SpiderException {
        PickerConfigurer configurer = spider.getPickerConfigurer();
        if (configurer == null) {
            return;
        }
        int maxDepth = context.getMaxDepth() - 1;
        @SuppressWarnings("unchecked")
        URLPicker<Resp> picker = (URLPicker<Resp>) configurer.getURLPicker();
        if (maxDepth >= 0 && picker != null) {
            PickerConfigurer.Matcher matcher = configurer.findMatcher(resp.getURL());
            if (matcher == null) {
                return;
            }
            List<URL> urlList = null;
            try {
                urlList = picker.pick(resp);
            } catch (Throwable e) {
            }
            if (spider.getFetchMode() == Consts.DEPTH_FIRST) {
                context.setPriority(context.getPriority() + 1);
            }
            if (urlList != null && urlList.size() > 0) {
                for (URL url : urlList) {
                    if (!matcher.isAccept(url)) {
                        continue;
                    }
                    Context sub = new Context(url);
                    sub.setMaxDepth(maxDepth);
                    sub.setPriority(context.getPriority());
                    sub.setExpire(context.getExpire());
                    spider.addFetchContext(sub);
                }
            }
        }
    }

    /**
     * 获取响应后交给Handler进行处理
     *
     * @param resp
     * @param context
     * @throws SpiderException
     */
    @SuppressWarnings("unchecked")
    private void handleResponse(Resp resp, Context context) throws SpiderException {

        for (ResponseHandler<Resp> responseHandler : spider.getGlobalResponseHandlerArray()) {
            responseHandler.onResponse(resp);
        }
        if (context.hasHandler()) {
            for (URLHandler<?, ?> handler : context.getHandlerList()) {
                ((URLHandler<Req, Resp>) handler).onResponse(resp);
            }
        }
    }

    /**
     * 异常时交给Handler进行处理
     *
     * @param e
     * @param context
     */
    private void handleException(Throwable e, Context context) {
        for (ExceptionHandler exceptionHandler : spider.getGlobalExceptionHandlerArray()) {
            exceptionHandler.onException(context, e);
        }
        if (context.hasHandler()) {
            for (URLHandler<?, ?> handler : context.getHandlerList()) {
                handler.onException(context, e);
            }
        }
    }

    /**
     * 处理URL
     *
     * @param context
     */
    private void processURL(Context context) {
        Connector<Spider<Req, Resp, Conn>, Req, Resp, Conn> connector = null;
        try {
            if (!context.hasHandler()) {
                handleFilter(context);
            }
            connector = createConnector(context.getURL().getProtocol());
            Req req = connector.createRequest(context.getURL());
            if (req == null) {
                throw new SpiderException("connector[" + connector + "] return null request");
            }
            handleRequest(req, context);
            Conn conn = openURLConnection(connector, req);
            Resp resp = connector.handleConnection(spider, req, conn);
            if (resp == null) {
                throw new SpiderException("connector[" + connector + "] return null response");
            }
            handleResponse(resp, context);
            handleURLPicker(resp, context);
        } catch (Throwable e) {
            handleException(e, context);
        } finally {
            if (connector != null) {
                try {
                    connector.destroy();
                } catch (Exception e) {
                }
            }
        }
    }

    public void run() {
        Context context;
        while ((context = waitForData(workerPool.getIdleTimeout())) != null) {
            try {
                long workInterval = workerPool.getWorkInterval();
                if (workInterval > 0) {
                    // spider speed control
                    synchronized (globalWorkerLock) {
                        long waitTime = workInterval + lastWorkTime - System.currentTimeMillis();
                        lastWorkTime = System.currentTimeMillis();
                        if (waitTime > 0) {
                            globalWorkerLock.wait(waitTime);
                        }
                    }
                }
                processURL(context);
            } catch (Throwable e) {
            } finally {
                synchronized (this) {
                    this.context = null;
                    this.notify();
                }
            }
        }
    }

    private Context waitForData(long timeout) {
        synchronized (this) {
            if (context != null) {
                return context;
            }
            try {
                workerPool.recycle(this);
                if (timeout <= 0) {
                    this.wait();
                } else {
                    this.wait(timeout);
                }
                isWorking = context != null;
                return context;
            } catch (Exception e) {
                return null;
            }
        }
    }

    protected void doWork(Context context) {
        if (context == null) {
            return;
        }
        synchronized (this) {
            if (this.context != null) {
                throw new IllegalStateException("worker is busy");
            }
            this.context = context;
            if (!isWorking) {
                isWorking = true;
                thread = new Thread(this, "SpiderWorkThread-" + workerCounter.incrementAndGet());
                thread.setDaemon(true);
                thread.start();
            } else {
                this.notify();
            }
        }
    }

    protected void waitDone(Context context, long timeout) throws SpiderException {
        synchronized (this) {
            if (context == this.context) {
                try {
                    if (timeout <= 0) {
                        this.wait();
                    } else {
                        this.wait(timeout);
                    }
                } catch (InterruptedException e) {
                    throw new SpiderException("operation is interrupted");
                }
            }
        }
    }

    protected boolean isBusy() {
        Thread thread = this.thread;
        return thread != null && thread.getState() == Thread.State.RUNNABLE;
    }

    protected boolean isAlive() {
        Thread thread = this.thread;
        return thread != null && thread.isAlive();
    }

    protected void destroy() {
        synchronized (this) {
            isWorking = false;
            context = null;
            if (thread != null) {
                thread.interrupt();
                thread = null;
            }
        }
    }
}